package com.atguigu.flink.streamapi.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * Created by Smexy on 2022/11/21
 * 
 *      1.14之后推荐使用  KafkaSource
 *
 *
 *      static <OUT> KafkaSourceBuilder<OUT> builder():
 *                  调用的时候，调用方式:   KafkaSource.<String>builder()
 */
public class Demo4_KafkaSource_New
{
    public static void main(String[] args) {
        
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建一个KafkaSource，就是一个消费者
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("hadoop102:9092")
            .setTopics("t1")
            .setGroupId("sz0704c")
            /*
                OffsetsInitializer:
                        OffsetsInitializer.earliest(): 最早的位置
                        OffsetsInitializer.latest(): 从最后的位置
                        OffsetsInitializer.committedOffsets(): 从提交的位置
                        OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST): 从提交的位置读取，如果从未提交过，由传入的策略决定
             */
            .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .setProperty("auto.commit.interval.ms","500")
            .setProperty("enable.auto.commit","true")
            .build();

        /*
            使用消费者读取数据
         */
        DataStreamSource<String> ds = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        ds.print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
        
    }
}
